package com.lvmama.rhino.analyze.processing

import com.lvmama.rhino.common.entity.{H5, NATIVE}
import com.lvmama.rhino.common.utils.JDBCUtil.Conversions
import com.lvmama.rhino.common.utils.Utils
import org.apache.spark.SparkContext
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions._
import org.apache.spark.storage.StorageLevel

/**
 * Created by SangXiaoduo on 2016/10/24.
 */
object WirelessFlowTransform {
  def process(sc: SparkContext, pageForward: DataFrame): Unit = {
    import com.lvmama.rhino.analyze.client.WirelessStat._

    val commons = Seq("deviceToken", "sessionId", "channelCode", "platformCode", "pageTypeCode", "timestamp", "pageParam", "buttonCode")

    //获取categoryId
    val getCategoryId = pageForward.select(commons.map(col): _*)
      .withColumn("category_id", when(col("pageParam").getItem("categoryId").isNull, "-1").otherwise(col("pageParam").getItem("categoryId")))

    //native页面访问事件
    val native = getCategoryId.filter(col("platformCode") =!= "2g53").coalesce(20)
      .withColumn("product_id", when(col("pageParam").getItem("productId").isNull, "-1").otherwise(col("pageParam").getItem("productId")))
      .withColumn("pageTotalCode", concat(col("pageTypeCode"), col("category_id"), col("product_id"), col("buttonCode")))
      .withColumn("pageTotalCodeLag", dataLag(col("pageTotalCode")))
      .withColumn("buttonLag", dataLag(col("buttonCode")))
      .filter(col("pageTotalCodeLag").isNull.or(col("pageTotalCode") !== col("pageTotalCodeLag")))
      .filter(col("buttonCode") !== "back")
      .filter(col("buttonLag").isNull.or(col("buttonLag") !== "back"))
      .drop(col("pageTotalCode"))
      .drop(col("pageTotalCodeLag"))
      .drop(col("buttonLag"))
      .drop(col("product_id"))
      .persist(StorageLevel.MEMORY_AND_DISK_SER)

    //h5页面访问事件
    val h5 = getCategoryId.filter(col("platformCode") === "2g53").coalesce(20)

    //总的访问量
    val pagePV = native.unionAll(h5).coalesce(20)

    val columns = Seq("sessionId", "channelCode", "platformCode", "pageTypeCode", "timestamp", "category_id", "land_code")
    val group = Seq("channelCode", "platformCode", "pageTypeCode", "land_code", "category_id")

    //session下用户访问的第一条日志
    val firstColumns = pagePV
      .select(Seq(col("sessionId"), col("timestamp"), col("pageTypeCode")): _*)
      .withColumn("first_columns", dataLag(col("timestamp")))
      .where(col("first_columns").isNull)
      .withColumn("land_code", col("pageTypeCode"))
      .select(col("sessionId").cast("string"), col("land_code"))

    //选择公用字段
    val pageData = pagePV.join(firstColumns, "sessionId")

    //session下用户访问的最后一条日志
    val lastColumns = pageData
      .select(columns.map(col(_)) ++ Seq(wireless_lead.as("last_columns")): _*)
      .where(col("last_columns").isNull)

    //页面访问次数
    val pageVisitCounts = pageData
      .groupBy(group.map(col(_)): _*)
      .agg(count("*").as("amount"),countDistinct("sessionId").as("amount_uv"))
      .select(col("channelCode").cast("string"), col("platformCode").cast("string"),
        col("pageTypeCode").cast("string"), col("land_code").cast("string"), col("category_id").cast("string"),
        col("amount"),col("amount_uv"))


    //页面退出次数
    val pageExitCounts = lastColumns
      .groupBy(group.map(col(_)): _*)
      .agg(count("*").as("exit_amount"))
      .select(col("channelCode").cast("string"), col("platformCode").cast("string"), col("pageTypeCode").cast("string"),
        col("land_code").cast("string"), col("category_id").cast("string"), col("exit_amount"))

    //获取当前时间的前一天日期
    val yesterday = Utils.getYesterday()
    import Conversions._

    //访问流量统计并将结果存入MySQL
    pageVisitCounts.join(pageExitCounts, group, "left_outer")
      .select(col("channelCode").as("channel_code"), col("platformCode").as("platform_code"),
        col("pageTypeCode").as("page_code"), col("land_code"),
        when(col("category_id") === "-1", null).otherwise(col("category_id")).as("category_id"),
        col("amount"), when(col("exit_amount").isNull, 0).otherwise(col("exit_amount")).as("exit_amount"),
        col("amount_uv"))
      .withColumn("oper_date", lit(yesterday))
        .coalesce(20)
      .insertDF2MysqlDirect("flow_statistics")

  }
}
